package com.burges.net.tableAPIAndSQL.table

import org.apache.flink.api.common.typeinfo.Types
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.TableEnvironment
import org.apache.flink.table.descriptors.{Kafka, Rowtime, Schema}

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/14 
  * 描述     外部链接器应用实例
  */
object KafkaTableConnectorDEMO {

	def main(args: Array[String]): Unit = {
		val environment = StreamExecutionEnvironment.getExecutionEnvironment
		// 获取TableEnvironment
		val tStreamEnv = TableEnvironment.getTableEnvironment(environment)
		tStreamEnv
				//指定需要连接的外部系统，以下指定KafkaConnector
        		.connect(
					new Kafka()
        			.version("0.10")
        			.topic("my-topic")
        			.startFromEarliest()
        			.property("zookeeper.connect", "localhost:2181")
        			.property("bootstap.servers", "localhost:9092")
				)
				//指定TableFormat
//        		.withFormat(
//					new Json()
//					.failOnMissingField(true)   // optional: flag whether to fail if a field is missing or not, false by default
//					// or by using a JSON schema which parses to DECIMAL and TIMESTAMP
//					.jsonSchema(
//						"{" +
//								"  type: 'object'," +
//								"  properties: {" +
//								"    lon: {" +
//								"      type: 'number'" +
//								"    }," +
//								"    rideTime: {" +
//								"      type: 'string'," +
//								"      format: 'date-time'" +
//								"    }" +
//								"  }" +
//								"}"
//					)
//				)
				//指定TableSchema
        		.withSchema(
			        new Schema()
        			        .field("id", Types.INT)
        			        .field("name", Types.STRING)
        			        .field("rowtime", Types.SQL_TIMESTAMP)
        			        .rowtime(
				                new Rowtime()
        				                .timestampsFromField("timestamp")
        				                .watermarksPeriodicBounded(60000)
			                )
		        )
				//指定数据更新模式为AppendMode
        		.inAppendMode()
				//注册TableSource，指定Table名称为KafkaInputTable
        		.registerTableSource("KafkaInputTable")
	}

}
